package com.gx.mvp.rx;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 可缓存的队列执行器。
 *
 * Created by Guoxin on 2019/12/05.
 */
public final class CachedQueueExecutor implements ExecutorService {

    private ExecutorService executor;
    private int             maxThreads = 8;
    private AtomicInteger   numThreads = new AtomicInteger(0);

    private ConcurrentLinkedQueue<Runnable> queue;

    /**
     * 构造函数。
     *
     * @param maxThreads 指定最大并发线程数量。
     */
    private CachedQueueExecutor(int maxThreads) {
        this.executor = Executors.newCachedThreadPool();
        this.maxThreads = maxThreads;
        this.queue = new ConcurrentLinkedQueue<Runnable>();
    }

    /**
     * 创建可缓存队列执行器。
     *
     * @param maxThreads 指定最大并发线程数量。
     *
     * @return
     */
    public static CachedQueueExecutor newCachedQueueThreadPool(int maxThreads) {
        if (maxThreads <= 0) {
            throw new IllegalArgumentException("Max threads is not less than zero.");
        }

        return new CachedQueueExecutor(maxThreads);
    }

    /**
     * 重置最大线程数。
     *
     * @param maxThreads 指定新的最大线程数量。
     */
    public void resetMaxThreads(int maxThreads) {
        this.maxThreads = maxThreads;
    }

    @Override
    public void execute(Runnable command) {
        // 命令入队
        this.queue.offer(command);

        if (this.numThreads.get() < this.maxThreads) {
            // 线程计数
            this.numThreads.incrementAndGet();

            this.executor.execute(new QueueTask());
        }
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return this.executor.awaitTermination(timeout, unit);
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
        return null;
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        return this.executor.invokeAll(tasks, timeout, unit);
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        return this.executor.invokeAny(tasks);
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return this.executor.invokeAny(tasks, timeout, unit);
    }

    @Override
    public boolean isShutdown() {
        return this.executor.isShutdown();
    }

    @Override
    public boolean isTerminated() {
        return this.executor.isTerminated();
    }

    @Override
    public void shutdown() {
        this.queue.clear();
        this.executor.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        this.queue.clear();
        return this.executor.shutdownNow();
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return this.executor.submit(task);
    }

    @Override
    public Future<?> submit(Runnable task) {
        return this.executor.submit(task);
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return this.executor.submit(task, result);
    }

    /**
     * 执行队列任务。
     */
    protected final class QueueTask implements Runnable {
        protected QueueTask() {
        }

        @Override
        public void run() {
            do {
                Runnable task = queue.poll();
                if (null != task) {
                    task.run();
                }
            }
            while (!queue.isEmpty());

            numThreads.decrementAndGet();
        }
    }
}
